%%% 
%%% a heng job module, used to calculate the word count
%%%
-module(wordcount). 
-compile(export_all).

-record(read_state, 
                {fd = 0,
                 data :: binary() = <<>>,
                 pos :: integer() = 0,
                 eof :: bool() = false,
                 liner :: binary(),
                 lineno :: integer()
                }).

 
job_config() ->
    [{name, "word count"},	% 任务名称
    {vsn, "1.0"},			% 版本
 	{author, "cheng"},      % 作者 	
 	{map_count, 4}, 	    % maper数目
 	{reduce_count, 2},     % reducer数目 	
 	{timeout, infinity}     % 超时时间无限 	
 	].
 
input_data() ->
 	[
		"http://hengproject.org/data1.txt",
 		"http://hengproject.org/data2.txt",
 		"http://hengproject.org/data3.txt",
        "http://hengproject.org/data4.txt"
 	].

map_init() ->
 	ok.
 
%% return: {{key, value}, state} | {[{key, value}], state}| eof
map_reader(File, none) ->    
    {ok, Fd} = file:open(File, [binary, raw, read]),
    map_reader(File, #read_state{fd = Fd});
map_reader(_File, #read_state{eof = true}) ->
    eof;
map_reader(File, State = #read_state{data = <<>>}) ->
    #read_state{fd = Fd,                 
                pos = Pos,
                liner = Remain,
                lineno = LineNo
               } = State,
    case map_reader1(Fd, Pos) of
        eof ->
            case Remain of
                <<>> ->
                    eof;
                _ ->
                    {{LineNo + 1, Remain}, State#read_state{eof = true}}
            end;
        Data2 ->
            Pos2 = Pos + byte_size(Data),
            map_reader(File, State#read_state{pos = Pos2, data = Data2})
    end;
map_reader(File, State = #read_state{}) ->
    #read_state{fd = Fd, 
                data = Data,
                pos = Pos,
                liner = Remain, 
                lineno = LineNo
               } = State,
               
    %% get line from data
    case io_lib:collect_line(start, Data, none) of
        {stop, Line, Rest} ->
            LineNo2 = LineNo + 1,                    
            Line2 = add_pre_lineremain(Line, Remain),                        
            State2 = State#read_state{data = Rest, lineno = LineNo2, liner = <<>>},
            {{LineNo2, Line2}, State2};
        [L] ->
            State2 = State#read_state{data = <<>>, liner = L},
            map_reader(File, State2)
    end.
    

map_reader1(Fd, Pos) ->
    case file:pread(Fd, Location, 1024000) of % 1M
        {ok, Data} ->
            Data;
        eof ->
            eof
    end.

add_pre_lineremain(Data, <<>>) ->
    Data;
add_pre_lineremain(Data, Remain) ->
    <<Remain/binary, Data/binary>>.
 
%% Map函数
map(LineNo, Line) -> 
 	[{W, 1} || W <- word_in_line(Line)]. 	
    
%% Map中间数据保存文件名: wordcount_inter
map_inter() ->
 	?MODULE ++ "_inter".
 
%% 必须
%% 对中间key进行空间划分,制定给对应的Reduer
%% 返回:对应的Reducer索引
%% 则中间文件为 : wordcount_inter_1, wordcount_inter_2 
partition(K, M) ->
 	erlang:md5(term_to_binary(K)) rem M.

%% 
reduce(Word, InterList) ->    
    {Word, lists:sum(InterList)}.
 
%% 将最终数据输出
output_data(Data, Master) ->
    ok.
